from pyspark import SparkConf, SparkContext
import os
import json

if __name__ == '__main__':
    os.environ["PYSPARK_PYTHON"] = "D:/Python/Python310/python.exe"
    conf = SparkConf().setMaster("local[*]").setAppName("create rdd")
    sc = SparkContext(conf=conf)
    file_rdd = sc.textFile("D:/sparkTest.txt")
    split_rdd = file_rdd.flatMap(lambda x: x.split("|"))
    json_rdd = split_rdd.map(lambda x: json.loads(x))
    print(f"json_rdd的值是：{json_rdd.collect()}")
    map_rdd1 = json_rdd.map(lambda x: (x['areaName'], int(x['money'])))
    map_rdd2 = map_rdd1.reduceByKey(lambda a, b: a + b)
    # 各个城市销售额排名，从大到小
    sort_rdd = map_rdd2.sortBy(lambda x: x[1], ascending=False, numPartitions=1)
    print(sort_rdd.collect())
    # 全部城市，有哪些商品类别在售卖
    map_rdd3 = json_rdd.map(lambda x: x['category'])
    print(map_rdd3.distinct().collect())
    # 北京市有哪些商品类别在售卖
    rdd_filter = json_rdd.filter(lambda x: x['areaName'] == '北京')
    map_rdd4 = rdd_filter.map(lambda x: x['category']).distinct()
    print("北京在售商品类别：", map_rdd4.collect())
